package canal

import (
	"bytes"
	"fmt"
	"io"
	"log"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"
)

const leakyBufSize = 4108 // data.len(2) + hmacsha1(10) + data(4096)
const maxNBuf = 2048

var (
	leakyBuf = NewLeakyBuf(maxNBuf, leakyBufSize)
	//leakyBufCommand = NewLeakyBufCommand(maxNBuf)

	//lazyCmdPool     = leakyBufCommand
	xmit = leakyBuf
)

func (c *Canal) dump(w io.Writer) error {
	conn := c.getNetConn()
	for {
		buf := xmit.Get()
		_, err := io.CopyBuffer(w, conn, buf)
		if err != nil {
			return err
		}
		xmit.Put(buf)
	}
}

func (c *Canal) dumpAndParse() (err error) {
	err = c.replconf()
	if err != nil {
		return err
	}
	r, w := io.Pipe()		// Pipe 创建一个同步内存管道。它可以用来连接代码期望一个 io.Reader 与代码期待一个 io.Writer 。

	done := make(chan error, 1)

	go func() {
		err := c.handler(r)		// 解码
		_ = w.CloseWithError(err)

		done <- err
	}()

	err = c.dump(w)
	_ = w.CloseWithError(err)
	err = <-done

	return err
}

func(c *Canal) replIdOffsetComper(Str []byte) (sig bool ){
	_cmd:=append(append([]byte{70,85,76,76,82,69,83,89,78,67,32}, []byte(c.replId)...))
	_cmd = append(_cmd,[]byte{32}...)
	_cmd = append(_cmd,[]byte(strconv.FormatInt(c.offset,10))...)
	sig  = bytes.EqualFold(Str,_cmd)
	return sig

}

func (c *Canal) handler(rd io.Reader) error {	// 获得值
	resp := newReader(rd)		// 返回一个Reader，用于读取值类型。
	one := sync.Once{}			// “Once”是将完全只执行一次一个动作的对象。
	for {
		select {
		case aerr := <-c.ackErrC:
			return aerr
		default:
		}

		val, n, err := resp.readBulk()		// val: 输出值	 readBulk: reads the next Value from Reader.
		if err != nil {
			return err
		}
		switch val.Typ {
		case '-', ':', '$':
		case '+':
			if bytes.HasPrefix(val.Str, []byte{70,85,76,76,82,69,83,89,78,67}) {	// +FULLRESYNC runid offset
				resp, err = decodeStream(resp, c)

				if err != nil {
					return err
				}

				// 安全匹配 replID 和 offset , 防止比串改
				sig := c.replIdOffsetComper(val.Str)

				if sig == false{
					log.Print("ReplID or Offset inconsistent !")
					os.Exit(0)
				}

			} else if bytes.HasPrefix(val.Str,[]byte{67,79,78,84,73,78,85,69}) {	// 收到 +CONTINUE	使用master replid 和 offset 功能恢复
				ss := strings.Split(val.String(), " ")
				if len(ss) != 2 {
					return fmt.Errorf("%s(%s)", "error CONTINUE resp", val.String())
				}
				c.replId = ss[1]
			}
		case '*':
			// cmd := lazyCmdPool.Get().(*Command)
			cmd := &Command{}
			cmd.Set(buildStrCommand(val.String())...)
			err = c.Command(cmd)
			if err != nil {
				return err
			}
			// lazyCmdPool.Put(cmd)
			c.Increment(int64(n))
		default:
			log.Printf("address %s replId %s unknown opcode %v size %d", c.ip, c.replId, val, val.Size)
		}

		one.Do(func() {
			//if sig == false{
			//	signalChan := make(chan os.Signal,1) //创建一个信号量的chan，缓存为1，（0,1）意义不大
			//	signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM)//让进城收集信号量。
			//	<-signalChan
			//	log.Println("replID & offset is corropt!")
			//}
			go c.replack()
		})

		fmt.Printf("val: %v,replId: %v, offset: %d\n",val.ArrayV,c.replId,c.offset)
	}

}

func (c *Canal) replack() {
	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-c.closeReplica:
			return
		default:
		}
		err := c.wr.writeMultiBulk("replconf", "ack", c.Offset())
		if err != nil {
			c.ackErrC <- err
		}
		<-ticker.C
	}
}
